package movee.zookeeper.leader.selector.cluster;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import movee.zookeeper.leader.selector.config.ZkClientConfig;
import movee.zookeeper.leader.selector.utils.ThreadUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;
import org.apache.curator.framework.recipes.leader.Participant;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;

/**
 * 分布式选主
 *
 * @author movee
 */
@Slf4j
@Component
public class ClusterNodeStateService implements CommandLineRunner, Closeable {

    private final ClusterEventListenerRegistry<ClusterNodeStateListener> listeners;

    private final Map<String, ExecutorService> listenerExecutors;

    private CuratorFramework curatorClient;

    private LeaderLatch leaderLatch;

    private ClusterNodeState clusterState = ClusterNodeState.INIT;

    private final Object lock = new Object();

    public ClusterNodeStateService(final ZkClientConfig config,
                                   @Value("${server.port}")
                                   Integer serverPort) {

        listeners = new SimpleClusterEventListenerRegistry<>();
        listenerExecutors = new HashMap<>();


        ThreadFactory threadFactory = ThreadUtils.namedThreadFactory("thread-curator-notify-%d");

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);

        curatorClient = CuratorFrameworkFactory.builder()
                .connectString(config.getConnectString())
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(config.getSessionTimeoutMs())
                .connectionTimeoutMs(config.getConnectionTimeoutMs())
                .threadFactory(threadFactory)
                .build();

        try {
            String id = InetAddress.getLocalHost().getHostName() + ":" + serverPort;
            leaderLatch = new LeaderLatch(curatorClient, config.getLeaderLatchPath(), id);

            leaderLatch.addListener(new ZkLeaderLatchListener());
        } catch (Exception e) {
            log.error(ExceptionUtils.getStackTrace(e));
        }
    }

    @Override
    public void run(String... strings) throws Exception {

        try {
            log.info("start ClusterNodeStateService......");
            curatorClient.start();

            leaderLatch.start();
        } catch (Exception e) {
            log.error(ExceptionUtils.getStackTrace(e));
        }

    }

    @Override
    public void close() throws IOException {

        log.info("close ClusterNodeStateService......");
        CloseableUtils.closeQuietly(leaderLatch);
        CloseableUtils.closeQuietly(curatorClient);

    }

    public ClusterInfo info() {

        String participant = leaderLatch.getId();
        List<String> participants = new ArrayList<>();
        String leader = "";
        boolean hasLeader = false;
        boolean isLeader = false;
        List<String> listenerNames = new ArrayList<>();
        try {
            participants = leaderLatch.getParticipants().stream().map(Participant::getId).collect(Collectors.toList());
            leader = leaderLatch.getLeader().getId();
            isLeader = Objects.equals(participant, leader);
            hasLeader = !StringUtils.isEmpty(leader);
            listenerNames = listeners.getListenerNames();
        } catch (Exception e) {
            log.error(ExceptionUtils.getStackTrace(e));
        }

        return ClusterInfo.builder().participant(participant)
                .participants(participants)
                .hasLeader(hasLeader)
                .isLeader(isLeader)
                .leader(leader)
                .listeners(listenerNames)
                .build();
    }

    public void registerClusterNodeStateListener(final String name, final ClusterNodeStateListener listener) {
        synchronized (lock) {
            log.info("register ClusterNodeStateListener, name: {}", name);
            listeners.registerClusterEventListener(name, listener);
            listenerExecutors.put(name, Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
                    .setNameFormat("thread-cluster-event-listener-" + listener.getName() + "-%d")
                    .build()));

            // 通知当前节点状态
            listenerExecutors.get(name)
                    .submit(() -> listener.onClusterNodeStateChange(ClusterNodeState.INIT, clusterState));

        }
    }

    public void removeClusterNodeStateListener(final String name) {
        synchronized (lock) {
            if (!StringUtils.isEmpty(name)) {
                log.info("delete ClusterNodeStateListener, name: {}", name);
                listeners.removeClusterEventListener(name);
                listenerExecutors.remove(name);
            }
        }
    }

    private class ZkLeaderLatchListener implements LeaderLatchListener  {

        @Override
        public void isLeader() {

            ClusterNodeState previous = clusterState;
            clusterState = ClusterNodeState.MASTER;

            log.info("ClusterNodeState changed, previous: {}, new: {}", previous, clusterState);
            if (previous != clusterState) {
                notify(previous, clusterState);
            }

        }

        @Override
        public void notLeader() {

            ClusterNodeState previous = clusterState;
            clusterState = ClusterNodeState.SLAVE;

            log.info("ClusterNodeState changed, previous: {}, new: {}", previous, clusterState);
            if (previous != clusterState) {
                notify(previous, clusterState);
            }
        }


        private void notify(final ClusterNodeState previous, final ClusterNodeState current) {

            synchronized (lock) {
                for (ClusterNodeStateListener listener : listeners) {
                    ExecutorService executorService = listenerExecutors.get(listener.getName());
                    if (executorService != null) {
                        executorService.submit(() -> listener.onClusterNodeStateChange(previous, current));
                    }

                }
            }

        }

    }

}
